package com.atguigu.day09;

import com.atguigu.util.UserBehavior;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;

public class Example5 {
    public static void main(String[] args) {
        var env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        var source = FileSource
                .forRecordStreamFormat(
                        // 按行消费
                        new TextLineInputFormat(),
                        new Path("/home/yuantuzhi/flinktutorial0926/src/main/resources/UserBehavior.csv")
                )
                .build();

        var stream = env
                .fromSource(source, WatermarkStrategy.noWatermarks(), "user-behavior")
                .flatMap(new StringToUserBehavior())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                                .withTimestampAssigner((element, recordTimestamp) -> element.ts)
                );

        // 获取表执行上下文环境
        var tableEnv = StreamTableEnvironment
                .create(
                        env,
                        EnvironmentSettings.newInstance().inStreamingMode().build()
                );

        // 将数据流转换成动态表
        var table = tableEnv
                .fromDataStream(
                        stream,
                        $("userId").as("user_id"),
                        $("productId").as("product_id"),
                        $("categoryId").as("category_id"),
                        $("type").as("behavior_type"),
                        $("ts").rowtime().as("event_time") // `.rowtime`表示这一列是事件时间
                );

        // 将动态表注册为临时视图
        tableEnv.createTemporaryView("user_behavior", table);

        // 滑动窗口：HOP(时间戳的列名，滑动距离，窗口长度)
        // 滚动窗口：TUMBLE(时间戳的列名，窗口长度)
        // 会话窗口：SESSION(时间戳的列名，超时时间)
        // 使用COUNT有OOM的风险，因为收集了窗口中的所有元素
        var innerSQL = "SELECT" +
                                "  product_id," +
                                "  COUNT(product_id) AS cnt," +
                                "  HOP_START(event_time, INTERVAL '5' MINUTES, INTERVAL '1' HOURS) AS window_start_time," +
                                "  HOP_END(event_time, INTERVAL '5' MINUTES, INTERVAL '1' HOURS) AS window_end_time" +
                                "  FROM user_behavior GROUP BY product_id, HOP(event_time, INTERVAL '5' MINUTES, INTERVAL '1' HOURS)";

        var midSQL = "SELECT *, ROW_NUMBER() OVER (PARTITION BY window_end_time ORDER BY cnt DESC) AS row_num FROM (" + innerSQL + ")";

        var outerSQL = "SELECT * FROM (" + midSQL + ") WHERE row_num <= 3";

        var resultTable = tableEnv.sqlQuery(outerSQL);

        tableEnv.toChangelogStream(resultTable).print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static class StringToUserBehavior implements FlatMapFunction<String, UserBehavior> {
        @Override
        public void flatMap(String in, Collector<UserBehavior> out) throws Exception {
            var fields = in.split(",");
            var userBehavior = new UserBehavior(
                    fields[0], fields[1], fields[2], fields[3],
                    Long.parseLong(fields[4]) * 1000L
            );

            if (userBehavior.type.equals("pv"))
                out.collect(userBehavior);
        }
    }
}

// +U[5051027, 3, 2017-11-26T00:05, 2017-11-26T01:05, 1]
// +U[3493253, 3, 2017-11-26T00:05, 2017-11-26T01:05, 2]
// +U[4261030, 3, 2017-11-26T00:05, 2017-11-26T01:05, 3]

// 第1名的商品id是：5051027,浏览次数是：3
// 第2名的商品id是：3493253,浏览次数是：3
// 第3名的商品id是：4261030,浏览次数是：3